package com.huwaiwai.threadpool;

import java.util.ArrayList;
import java.util.HashSet;
import java.util.List;
import java.util.concurrent.ArrayBlockingQueue;
import java.util.concurrent.BlockingQueue;
import java.util.concurrent.RejectedExecutionHandler;
import java.util.concurrent.ThreadFactory;
import java.util.concurrent.ThreadPoolExecutor;
import java.util.concurrent.TimeUnit;
import java.util.concurrent.atomic.AtomicInteger;
import java.util.concurrent.locks.AbstractQueuedSynchronizer;
import java.util.concurrent.locks.Condition;
import java.util.concurrent.locks.ReentrantLock;

/**
 * Created by hww on 2018/7/7 上午12:17.
 */
public class MyThreadPool extends ThreadPoolExecutor {

    private final AtomicInteger ctl = new AtomicInteger(ctlOf(RUNNING, 0));

    /** 32位int类型的前三位保存线程池的状态 */
    private static final int COUNT_BITS = Integer.SIZE - 3;
    /** 32位int类型的后29位表示线程池的容量 */
    private static final int CAPACITY   = (1 << COUNT_BITS) - 1;

    // 接收新任务    处理已经进入队列的任务
    private static final int RUNNING = -1 << COUNT_BITS;
    // 不接收新任务  处理已经进入队列的任务
    private static final int SHUTDOWN = 0 << COUNT_BITS;
    // 不接收新任务  不处理已经进入队列的任务  并且中断正在执行的任务
    private static final int STOP = 1 << COUNT_BITS;
    // 所有任务完成执行 workerCount = 0 线程转入当前状态后会执行terminated()方法
    private static final int TIDYING = 2 << COUNT_BITS;
    // terminated()已经执行完成
    private static final int TERMINATED = 3 << COUNT_BITS;

    public MyThreadPool(int corePoolSize, int maximumPoolSize, long keepAliveTime, TimeUnit unit, BlockingQueue<Runnable> workQueue) {
        super(corePoolSize, maximumPoolSize, keepAliveTime, unit, workQueue);
    }

    public MyThreadPool(int corePoolSize, int maximumPoolSize, long keepAliveTime, TimeUnit unit, BlockingQueue<Runnable> workQueue, ThreadFactory threadFactory) {
        super(corePoolSize, maximumPoolSize, keepAliveTime, unit, workQueue, threadFactory);
    }

    public MyThreadPool(int corePoolSize, int maximumPoolSize, long keepAliveTime, TimeUnit unit, BlockingQueue<Runnable> workQueue, RejectedExecutionHandler handler) {
        super(corePoolSize, maximumPoolSize, keepAliveTime, unit, workQueue, handler);
    }

    public MyThreadPool(int corePoolSize, int maximumPoolSize, long keepAliveTime, TimeUnit unit, BlockingQueue<Runnable> workQueue, ThreadFactory threadFactory, RejectedExecutionHandler handler) {
        super(corePoolSize, maximumPoolSize, keepAliveTime, unit, workQueue, threadFactory, handler);
    }

    //获取线程池状态
    private static int runStateOf(int c) {
        return c & ~CAPACITY;
    }
    //获取线程池中的工作线程的数量
    private static int workerCountOf(int c) {
        return c & CAPACITY;
    }

    //todo 方法含义待补充
    private static int ctlOf(int rs, int wc) {
        return rs | wc;
    }


    private static boolean runStateLessThan(int c, int s) {
        return c < s;
    }

    private static boolean runStateAtLeast(int c, int s) {
        return c >= s;
    }

    //是否是运行状态
    private static boolean isRunning(int c) {
        return c < SHUTDOWN;
    }

    //加一个工作线程
    private boolean compareAndIncrementWorkerCount(int expect) {
        return ctl.compareAndSet(expect, expect + 1);
    }

    //减一个工作线程
    private boolean compareAndDecrementWorkerCount(int expect) {
        return ctl.compareAndSet(expect, expect - 1);
    }

    //无线循环减少一个工作线程  直到成功
    private void decrementWorkerCount() {
        do {} while (! compareAndDecrementWorkerCount(ctl.get()));
    }

    //工作任务队列
    private final BlockingQueue<Runnable> workQueue = new ArrayBlockingQueue<>(1);

    /**
     * 原文中大致介绍了一下此处使用锁而不使用并发集合的取舍
     */
    private final ReentrantLock mainLock = new ReentrantLock();

    //保存了线程池的工作线程  只有获得主锁的时候才能操作
    private final HashSet<Worker> workers = new HashSet<>();

    //支持等待终止的条件
    private final Condition termination = mainLock.newCondition();

    //最大线程数
    private int largestPoolSize;

    //完成的任务数
    private long completedTaskCount;

    //线程工厂  创建线程池中的线程
    private volatile ThreadFactory threadFactory;

    //当线程池饱和时或停止时调用  拒绝策略
    private volatile RejectedExecutionHandler handler;

    //空闲线程存活时间
    private volatile long keepAliveTime;

    //是否允许核心线程超时关闭
    private volatile boolean allowCoreThreadTimeOut;

    //核心工作线程容量
    private volatile int corePoolSize;

    //线程池最大值
    private volatile int maximumPoolSize;

    //默认拒绝策略
    private static final RejectedExecutionHandler defaultHandler = new ThreadPoolExecutor.AbortPolicy();

    //将线程转换为停止状态的许可
    private static final RuntimePermission shutdownPerm = new RuntimePermission("modifyThread");

    //执行终止器的上下文  或者为空
//    private final AccessControlContext acc;


    //返回用于创建线程的线程工厂
    public ThreadFactory getThreadFactory() {
        return threadFactory;
    }

    private final class Worker extends AbstractQueuedSynchronizer implements Runnable {
        private static final long serialVersionUID = -8551132129066071494L;

        /** 执行任务的线程 如果线程工厂创建失败 则为空 */
        final Thread thread;
        /** 初始化第一个任务 可能为空 */
        Runnable firstTask;
        /** 线程完成的任务计数器 */
        volatile long completedTasks;

        Worker(Runnable firstTask) {
            setState(-1);
            this.firstTask = firstTask;
            this.thread = getThreadFactory().newThread(this);
        }

        public void run() {
            runWorker(this);
            //todo 此处应该是出发工作线程的run方法 稍后补充
        }

        protected boolean isHeldExclusively() {
            return getState() != 0;
        }

        protected boolean tryAcquire(int unused) {
            if (compareAndSetState(0, 1)) {
                setExclusiveOwnerThread(Thread.currentThread());
                return true;
            }
            return false;
        }

        protected boolean tryRelease(int unused) {
            setExclusiveOwnerThread(null);
            setState(0);
            return true;
        }

        public void lock() {
            acquire(1);
        }

        public boolean tryLock() {
            return tryAcquire(1);
        }

        public void unlock() {
            release(1);
        }

        public boolean isLocked() {
            return isHeldExclusively();
        }

        void interruptIfStarted() {
            Thread t;
            if (getState() >= 0 &&  (t = thread) != null && !t.isInterrupted()) {
                try {
                    t.interrupt();
                } catch (SecurityException ignore) {

                }
            }
        }
    }

    //尝试 线程池变为结束
    final void tryTerminate() {
        for (;;) {
            int c = ctl.get();
            if (isRunning(c) || runStateAtLeast(c, TIDYING) || (runStateOf(c) == SHUTDOWN && !workQueue.isEmpty())) {
                return;
            }
            if (workerCountOf(c) != 0) {
                interruptIdleWorkers(ONLY_ONE);
                return;
            }

            final ReentrantLock mainLock = this.mainLock;
            mainLock.lock();
            try {
                if (ctl.compareAndSet(c, ctlOf(TIDYING, 0))) {
                    try {
                        terminated();
                    } finally {
                        ctl.set(ctlOf(TERMINATED, 0));
                        termination.signalAll();
                    }
                    return;
                }
            } finally {
                mainLock.unlock();
            }
            //cas转换状态失败 重复循环
        }
    }

    //校验是否可以中断
    private void checkShutdownAccess() {
        SecurityManager security = System.getSecurityManager();
        if (security != null) {
            security.checkPermission(shutdownPerm);
            final ReentrantLock mainLock = this.mainLock;
            mainLock.lock();
            try {
                for (Worker w : workers) {
                    security.checkAccess(w.thread);
                }
            } finally {
                mainLock.unlock();
            }
        }
    }

    //中断工作线程
    private void interruptWorkers() {
        final ReentrantLock mainLock = this.mainLock;
        mainLock.lock();
        try {
            for (Worker w : workers)
                w.interruptIfStarted();
        } finally {
            mainLock.unlock();
        }
    }

    //中断等待任务的工作线程
    private void interruptIdleWorkers(boolean onlyOne) {
        final ReentrantLock mainLock = this.mainLock;
        mainLock.lock();
        try {
            for (Worker w : workers) {
                Thread t = w.thread;
                if (!t.isInterrupted() && w.tryLock()) {
                    try {
                        t.interrupt();
                    } catch (SecurityException ignore) {

                    } finally {
                        w.unlock();
                    }
                }
                if (onlyOne) {
                    break;
                }
            }
        } finally {
            mainLock.unlock();
        }
    }

    private void interuptIdleWorkers() {
        interruptIdleWorkers(false);
    }

    private static final boolean ONLY_ONE = true;

    final void reject(Runnable command) {
        handler.rejectedExecution(command, this);
    }

    void onShutdown() {

    }

    // 是否是运行或终止状态
    final boolean isRunningOrShutdown(boolean shutdownOK) {
        int rs = runStateOf(ctl.get());
        return rs == RUNNING || (rs == SHUTDOWN && shutdownOK);
    }

    //重排序队列
    private List<Runnable> drainQueue() {
        BlockingQueue<Runnable> q = workQueue;
        ArrayList<Runnable> taskList = new ArrayList<>();
        //此方法会把 队列中的元素转移到list中， 但是对于delayQueue不会 需要一个一个操作
        q.drainTo(taskList);
        if (!q.isEmpty()) {
            for (Runnable r : q.toArray(new Runnable[0])) {
                if (q.remove(r)) {
                    taskList.add(r);
                }
            }
        }
        return taskList;
    }

    /**
     * 增加一个工作线程
     * @param firstTask
     * @param core
     * @return
     */
    private boolean addWorker(Runnable firstTask, boolean core) {
        retry:
        for (;;) {
            int c = ctl.get();
            int rs = runStateOf(c);
            //校验线程池是否是停止状态
            if (rs >= SHUTDOWN && !(rs == SHUTDOWN && firstTask == null && !workQueue.isEmpty())) {
                //1.线程池处于STOP(停止)、TIDYING(整理)、TERMINATED(结束)状态
                //2.线程池处于SHUTDOWN(关闭)状态(不接受新任务,继续处理队列中任务),但是传入新任务时候
                //3.线程池处于SHUTDOWN(关闭)状态(不接受新任务,继续处理队列中任务),队列为空时
                return false;
            }
            //到此处说明线程池状态没问题
            for (;;) {
                int wc = workerCountOf(c);
                if (wc >= CAPACITY || wc >= (core ? corePoolSize : maximumPoolSize)) {
                    //1.当前线程池中的线程数大于等于线程池理论最大值(int字段的29位表示的最大数)
                    //2.方法入参core=true时,用核心线程数进行比对，否则用线程池最大值(初始化时设置的最大值)进行比对
                    return false;
                }
                if (compareAndIncrementWorkerCount(c)) {
                    //增加工作线程数成功 直接退出外层循环
                    break retry;
                }
                c = ctl.get();
                if (runStateOf(c) != rs) {
                    //若线程池状态与之前状态发生了变化 则退出到外层重新循环
                    continue retry;
                }
                //若线程池状态与之前状态未发生变化 则循环执行增加工作线程数方法
            }
        }

        boolean workerStarted = false;
        boolean workerAdded = false;

        Worker w = null;
        try {
            //创建一个工人  内部封装了一个用线程工厂创建的线程
            w = new Worker(firstTask);
            final Thread t = w.thread;
            //如果线程工厂创建的线程为空 则直接退出方法
            if (t != null) {
                final ReentrantLock mainLock = this.mainLock;
                mainLock.lock();
                try {
                    int rs = runStateOf(ctl.get());
                    //线程池处于RUNNING状态  或者  SHUTDOWN状态且无新任务需要入队(方法入参firstTask为空)
                    if (rs < SHUTDOWN || (rs == SHUTDOWN && firstTask == null)) {
                        if (t.isAlive()) {
                            //此处返回true标识线程已经在运行了
                            throw new IllegalThreadStateException();
                        }
                        //将工作者接入到工作者队列中
                        workers.add(w);
                        int s = workers.size();
                        //更新最大线程数
                        if (s > largestPoolSize) {
                            largestPoolSize = s;
                        }
                        workerAdded = true;
                    }
                } finally {
                    mainLock.unlock();
                }
                if (workerAdded) {
                    //启动线程
                    t.start();
                    workerStarted = true;
                }
            }
        } finally {
            if (!workerStarted) {
                //若上面未执行到启动线程步骤  执行增加工作者失败步骤
                addWorkerFailed(w);
            }
        }
        return workerStarted;
    }

    //增加工作者失败时 后续处理
    //清除队列中的无效工作者
    //减少增加的工作者计数
    private void addWorkerFailed(Worker w) {
        final ReentrantLock mainLock = this.mainLock;
        mainLock.lock();
        try {
            if (w != null) {
                workers.remove(w);
            }
            decrementWorkerCount();
            tryTerminate();
        } finally {
            mainLock.unlock();
        }
    }

    //清除超时的线程
    private void processWorkerExit(Worker w, boolean completedAbruptly) {
        if (completedAbruptly) {
            decrementWorkerCount();
        }

        //清除工作者
        final ReentrantLock mainLock = this.mainLock;
        mainLock.lock();
        try {
            completedTaskCount += w.completedTasks;
            workers.remove(w);
        } finally {
            mainLock.unlock();
        }

        tryTerminate();

        int c = ctl.get();
        if (runStateLessThan(c, STOP)) {
            //若当前线程池状态不是停止状态
            if (!completedAbruptly) {
                //不是由于抛出异常而清除工作者的情况 判断最小应该保留的线程数
                int min = allowCoreThreadTimeOut ? 0 : corePoolSize;
                if (min == 0 && !workQueue.isEmpty()) {
                    //若任务队列中不为空  则至少保留一个工作者
                    min = 1;
                }
                if (workerCountOf(c) >= min) {
                    return;
                }
            }
            //若当前工作者小于最屌保留工作者数 创建一个工作者
            addWorker(null, false);
        }
    }

    //获取一个任务
    private Runnable getTask() {
        boolean timedOut = false;
        for (;;) {
            int c = ctl.get();
            int rs = runStateOf(c);
            if (rs >= SHUTDOWN && (rs >= STOP || workQueue.isEmpty())) {
                //若线程池已关闭且任务列表为空，或者线程池已停止
                decrementWorkerCount();
                return null;
            }
            int wc = workerCountOf(c);
            boolean timed = allowCoreThreadTimeOut || wc > corePoolSize;

            if ((wc > maximumPoolSize || (timed && timedOut)) && (wc > 1 || workQueue.isEmpty())) {
                if (compareAndDecrementWorkerCount(c)) {
                    return null;
                }
                continue;
            }

            try {
                Runnable r = timed
                        ? workQueue.poll(keepAliveTime, TimeUnit.NANOSECONDS)
                        : workQueue.take();
                if (r != null) {
                    return r;
                }
                timedOut = true;
            } catch (InterruptedException retry) {
                timedOut = false;
            }
        }
    }

    final void runWorker(Worker w) {
        Thread wt = Thread.currentThread();
        Runnable task = w.firstTask;
        w.firstTask = null;
        w.unlock();
        boolean completedAbruply = true;

        try {
            while (task != null || (task = getTask()) != null) {
                w.lock();
                if ((runStateAtLeast(ctl.get(), STOP) || (Thread.interrupted() && runStateAtLeast(ctl.get(), STOP))) && !wt.isInterrupted()) {
                    wt.interrupt();
                }

                try {
                    beforeExecute(wt, task);
                    Throwable thrown = null;
                    try {
                        task.run();
                    } catch (RuntimeException x) {
                        thrown = x;
                        throw x;
                    } catch (Error x) {
                        thrown = x; throw x;
                    } catch (Throwable x) {
                        thrown = x; throw new Error(x);
                    } finally {
                        afterExecute(task, thrown);
                    }
                } finally {
                    task = null;
                    w.completedTasks++;
                    w.unlock();
                }
            }
            completedAbruply = false;
        } finally {
            processWorkerExit(w, completedAbruply);
        }
    }

    @Override
    public void execute(Runnable command) {
        if (command == null) {
            throw new NullPointerException();
        }
        int c = ctl.get();
        if (workerCountOf(c) < corePoolSize) {
            //若当前活动线程数小于核心线程数 尝试创建线程 并把当前任务作为线程的第一个任务
            if (addWorker(command, true)) {
                //创建成功 返回
                return;
            }
            //创建失败 重新获取线程池状态
            c = ctl.get();
        }
        //若线程池是运行状态 则尝试将任务添加到任务队列中
        if (isRunning(c) && workQueue.offer(command)) {
            //添加成功后再次校验线程池状态
            int recheck = ctl.get();
            if (!isRunning(recheck) && remove(command)) {
                //若线程处于非运行状态 则删除队列中的当前任务 并执行拒绝策略
                reject(command);
            } else if (workerCountOf(recheck) == 0) {
                //若当前工作线程数为0，则创建一个工作线程
                addWorker(null, false);
            }
        } else if (!addWorker(command, false)) {
            //若当前线程非运行状态或任务入队失败 则尝试创建一个新线程 若创建失败 则执行拒绝策略
            reject(command);
        }
    }


    @Override
    public void shutdown() {
        final ReentrantLock mainLock = this.mainLock;
        mainLock.lock();
        try {
            checkShutdownAccess();
            advanceRunState(SHUTDOWN);
            interruptIdleWorkers();
            onShutdown();
        } finally {
            mainLock.unlock();
        }
        tryTerminate();
    }

    private void interruptIdleWorkers() {
        interruptIdleWorkers(false);
    }

    private void advanceRunState(int targetState) {
        for (;;) {
            int c = ctl.get();
            if (runStateAtLeast(c, targetState) ||
                    ctl.compareAndSet(c, ctlOf(targetState, workerCountOf(c))))
                break;
        }
    }

    @Override
    public List<Runnable> shutdownNow() {
        List<Runnable> tasks;
        final ReentrantLock mainLock = this.mainLock;
        mainLock.lock();
        try {
            checkShutdownAccess();
            advanceRunState(STOP);
            interruptWorkers();
            tasks = drainQueue();
        } finally {
            mainLock.unlock();
        }
        tryTerminate();
        return tasks;
    }

    @Override
    public boolean isShutdown() {
        return !isRunning(ctl.get());
    }

    public boolean isTerminating() {
        int c = ctl.get();
        return ! isRunning(c) && runStateLessThan(c, TERMINATED);
    }

    @Override
    public boolean isTerminated() {
        return runStateAtLeast(ctl.get(), TERMINATED);
    }

    @Override
    public boolean awaitTermination(long timeout, TimeUnit unit) throws InterruptedException {
        long nanos = unit.toNanos(timeout);
        final ReentrantLock mainLock = this.mainLock;
        mainLock.lock();
        try {
            for (;;) {
                if (runStateAtLeast(ctl.get(), TERMINATED))
                    return true;
                if (nanos <= 0)
                    return false;
                nanos = termination.awaitNanos(nanos);
            }
        } finally {
            mainLock.unlock();
        }
    }
}
